package cn.whuc.spark

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo_CategoryTop10_1 {
  def main(args: Array[String]): Unit = {
    // 1 创建sparkContext
    val sc: SparkContext = new SparkContext(
      new SparkConf()
        .setMaster("local[*]")
        .setAppName(" ")
    )

    // 2 编写代码
    // 2019-07-17_95_26070e87-1ad7-49a3-8fb3-cc741facaddf_37_2019-07-17 00:00:02_手机_-1_-1_null_null_null_null_3
    //2019-07-17_95_26070e87-1ad7-49a3-8fb3-cc741facaddf_48_2019-07-17 00:00:10_null_16_98_null_null_null_null_19

    val datas: RDD[String] = sc.textFile("input/user_visit_action.txt")
    /*
      一行一种行为
      每个品类 点击 下单 支付
      思路1
        品类 点击次数
        品类 下单次数
        品类 支付次数
      想办法 编程
        (品类 (点击 下单 支付))
      排序
      前10
      ok
     */
    // 点击行为统计
    val clickRDD: RDD[(String, Int)] = datas.filter(line => {
      val strings: Array[String] = line.split("_")
      strings(6) != "-1"
    }).map(line => {
      val strings: Array[String] = line.split("_")
      (strings(6), 1)
    }).reduceByKey(_ + _)

    // 下单行为统计
    val orderRDD: RDD[(String, Int)] = datas.filter(line => {
      val strings: Array[String] = line.split("_")
      strings(8) != "null"
    }).flatMap(line => {
      val strings: Array[String] = line.split("_")
      val ids: Array[String] = strings(8).split(",")
      ids.map(id => {
        (id, 1)
      })
    }).reduceByKey(_+_)

    // 支付行为统计
    val payRDD: RDD[(String, Int)] = datas.filter(line => {
      val strings: Array[String] = line.split("_")
      strings(10) != "null"
    }).flatMap(line => {
      val strings: Array[String] = line.split("_")
      val ids: Array[String] = strings(10).split(",")
      ids.map(id => {
        (id, 1)
      })
    }).reduceByKey(_+_)

    /*
      现在有了
      (品类，点击次数)
      (品类，下单次数)
      (品类，支付次数)
      怎么处理？
      百度。。。
     */
    val result: Array[(String, (Int, Int, Int))] = clickRDD.cogroup(orderRDD, payRDD).mapValues {
      case (a, b, c) => {
        var a1 = 0
        var b1 = 0
        var c1 = 0
        if (a.iterator.hasNext) {
          a1 = a.iterator.next()
        }
        if (b.iterator.hasNext) {
          b1 = b.iterator.next()
        }
        if (c.iterator.hasNext) {
          c1 = c.iterator.next()
        }
        (a1, b1, c1)
      }
    }.sortBy(_._2, false).take(10)


    result.foreach(println)



    // 3 关闭上下文对象
    sc.stop()
  }

}
